package com.dtwave.cfstreaming.utils;
import com.dtwave.cfstreaming.bean.ResultData;
import com.dtwave.param.EsparamObj;
import org.apache.commons.lang.StringUtils;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.Requests;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.reindex.UpdateByQueryRequest;
import org.elasticsearch.script.Script;
import org.elasticsearch.script.ScriptType;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.Map;

public class ElasticUtils {
    private static final Logger LOGGER= LoggerFactory.getLogger(ElasticUtils.class);
    /**
     * 针对某个对象创建索引并返回indexRequest，在fink-sink过程中使用
     */
    public static IndexRequest createIndexAndQuery(RestHighLevelClient client, EsparamObj esparam, ResultData res){
        String indexName = "testIndex";
        //是否通过es反查，配置参数决定，不传入默认为false，存在id则不写入数据
        if(esparam.getIsQuery()){
            //反查代码
            return null;
        }
        Map<String, Object> json = getJson(res);
        //返回组装好的request
        return Requests.indexRequest().index(indexName).id(res.getPassword()).source(json);
    }

    private static Map<String,Object> getJson(ResultData res) {
        HashMap<String,Object> json = new HashMap<>();
        json.put("name",res.getName());
        json.put("age",res.getAge());
        json.put("password",res.getPassword());
        return json;
    }

    /**
     * 工具类依据指定字段对es更新
     */
    public static void updateStatus(RestHighLevelClient client,String key){
        String index = "testIndex";
        UpdateByQueryRequest updateRequest = new UpdateByQueryRequest(index);
        updateRequest.setConflicts("proceed");
        updateRequest.setRefresh(true);
        //条件
        BoolQueryBuilder must = QueryBuilders.boolQuery()
                .must(QueryBuilders.termQuery("name", key));
        try {
            updateRequest.setQuery(must);
            Map<String,Object> map = new HashMap<>();
        //需要被更新上的数据
            map.put("key","value");
        //以字符流的方式组装更新需要的es的script
            StringBuilder sb = new StringBuilder();
        //将map转成流
            map.keySet().stream().forEach(t->{
                sb.append("csx.source.").append(t).append("=params.").append(t).append(";");
            });
            updateRequest.setScript(new Script(ScriptType.INLINE,"painless",sb.toString(),map));
        //提交
            client.updateByQuery(updateRequest, RequestOptions.DEFAULT);
            LOGGER.debug("数据更新成功");
        }catch (Exception e){
            LOGGER.error("es数据更新失败");
        }
    }

    /**
     * 依据id查询数据
     */
    public static String getValueById(RestHighLevelClient client,String id){
        try {
            GetRequest getRequest = new GetRequest("testIndex");
            getRequest.id(id);
            GetResponse document = client.get(getRequest, RequestOptions.DEFAULT);
            if(document != null && document.isExists()){
                Map<String, Object> map = document.getSourceAsMap();
                //获取数据
                String key = (String)map.get("key");
                if(StringUtils.isNotBlank(key)){
                    return key;
                }
            }
        }catch (Exception e){
            LOGGER.error("查询出错");
        }
        return null;
    }


    /**
     * 依据索引中的内容查询返回数据
     */
    public static String getValueByValue(RestHighLevelClient client,String id) {
        String indexName  = "testIndex";
        try {
            SearchSourceBuilder builder = new SearchSourceBuilder();
            //获取一条数据
            builder.query(QueryBuilders.termQuery("key","value"))
                    .from(0).size(1);
            //获取多条数据
            //builder.query(QueryBuilders.termQuery("key","value"))
            SearchRequest searchRequest = new SearchRequest(indexName);
            searchRequest.source(builder);
            SearchResponse search = client.search(searchRequest, RequestOptions.DEFAULT);
            if(search != null && search.getHits()!= null){
                SearchHit[] hits = search.getHits().getHits();
                if(hits != null && hits.length >0){
                    for (SearchHit hit : hits) {
                        Map<String, Object> sourceAsMap = hit.getSourceAsMap();
                        return sourceAsMap.get("word").toString();
                    }
                }
            }
        }catch (Exception e){
            LOGGER.error("数据获取失败！");
        }
        return null;
    }
}
